/*
 *
 *  * Copyright (c) iwindplus Technologies Co., Ltd.2024-2030, All rights reserved.
 *
 *
 */

package com.iwindplus.base.kafka;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.iwindplus.base.domain.constant.CommonConstant.SymbolConstant;
import com.iwindplus.base.kafka.domain.property.KafkaProperty;
import com.iwindplus.base.kafka.service.KafkaDynamicRegistry;
import jakarta.annotation.Resource;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.core.reactive.ReactiveKafkaConsumerTemplate;
import org.springframework.kafka.core.reactive.ReactiveKafkaProducerTemplate;
import org.springframework.kafka.support.serializer.JsonDeserializer;
import org.springframework.kafka.support.serializer.JsonSerializer;
import reactor.kafka.receiver.ReceiverOptions;
import reactor.kafka.sender.SenderOptions;

/**
 * Kafka配置.
 *
 * @author zengdegui
 * @since 2020/4/28
 */
@Slf4j
@Configuration
@ImportAutoConfiguration({KafkaAutoConfiguration.class})
@ConditionalOnProperty(prefix = "spring.kafka", name = "enabled", havingValue = "true", matchIfMissing = true)
@EnableConfigurationProperties(KafkaProperty.class)
public class KafkaConfiguration {

    @Resource
    private KafkaProperty property;

    @Resource
    private KafkaProperties kafkaProperties;

    @Resource
    private ObjectMapper objectMapper;

    @Resource
    private KafkaAdmin kafkaAdmin;

    /**
     * 创建AdminClient.
     *
     * @return AdminClient
     */
    @ConditionalOnMissingBean
    @Bean
    public AdminClient adminClient() {
        AdminClient adminClient = AdminClient.create(kafkaAdmin.getConfigurationProperties());
        log.info("AdminClient={}", adminClient);
        return adminClient;
    }

    /**
     * 动态创建主题初始化器.
     *
     * @param adminClient adminClient
     * @return kafkaDynamicRegistry
     */
    @Bean
    public KafkaDynamicRegistry kafkaDynamicRegistry(AdminClient adminClient) {
        final KafkaDynamicRegistry kafkaDynamicRegistry = new KafkaDynamicRegistry(adminClient, this.property);
        log.info("KafkaDynamicRegistry={}", kafkaDynamicRegistry);
        return kafkaDynamicRegistry;
    }

    /**
     * 创建生产者工厂.
     *
     * @param valueSerializer 序列化
     * @return ProducerFactory<?, ?>
     */
    @Primary
    @Bean
    public ProducerFactory<?, ?> producerFactory(JsonSerializer<Object> valueSerializer) {
        final StringSerializer keySerializer = new StringSerializer();
        Map<String, Object> props = kafkaProperties.buildProducerProperties();
        final DefaultKafkaProducerFactory<String, Object> producerFactory = new DefaultKafkaProducerFactory<>(props,
            keySerializer, valueSerializer);
        producerFactory.setValueSerializer(valueSerializer);
        log.info("ProducerFactory<?, ?>={}", producerFactory);
        return producerFactory;
    }

    /**
     * 创建发送者选项.
     *
     * @param valueSerializer 值序列化
     * @return SenderOptions<String, Object>
     */
    @Primary
    @Bean
    public SenderOptions<String, Object> senderOptions(JsonSerializer<Object> valueSerializer) {
        final StringSerializer keySerializer = new StringSerializer();
        Map<String, Object> props = kafkaProperties.buildProducerProperties();
        final SenderOptions<String, Object> options = SenderOptions.<String, Object>create(props)
            .withKeySerializer(keySerializer)
            .withValueSerializer(valueSerializer);
        log.info("SenderOptions<String, Object>={}", options);
        return options;
    }

    /**
     * 创建 ReactiveKafkaProducerTemplate
     *
     * @param senderOptions
     * @return ReactiveKafkaProducerTemplate<String, Object>
     */
    @Bean
    public ReactiveKafkaProducerTemplate<String, Object> reactiveKafkaProducerTemplate(SenderOptions<String, Object> senderOptions) {
        final ReactiveKafkaProducerTemplate<String, Object> reactiveKafkaProducerTemplate = new ReactiveKafkaProducerTemplate<>(
            senderOptions);
        log.info("ReactiveKafkaProducerTemplate<String, Object>={}", reactiveKafkaProducerTemplate);
        return reactiveKafkaProducerTemplate;
    }

    /**
     * 创建消费者工厂.
     *
     * @param valueDeserializer 反序列化
     * @return ConsumerFactory<?, ?>
     */
    @Primary
    @Bean
    public ConsumerFactory<?, ?> consumerFactory(JsonDeserializer<Object> valueDeserializer) {
        final StringDeserializer keyDeserializer = new StringDeserializer();
        Map<String, Object> props = kafkaProperties.buildConsumerProperties();
        final DefaultKafkaConsumerFactory<String, Object> consumerFactory = new DefaultKafkaConsumerFactory<>(props,
            keyDeserializer, valueDeserializer);
        log.info("ConsumerFactory<?, ?>={}", consumerFactory);
        return consumerFactory;
    }

    /**
     * 创建接收者选项.
     *
     * @param valueDeserializer 反序列化
     * @return ReceiverOptions<String, Object>
     */
    @Primary
    @Bean
    public ReceiverOptions<String, Object> receiverOptions(JsonDeserializer<Object> valueDeserializer) {
        Map<String, Object> props = kafkaProperties.buildConsumerProperties();
        // 清掉会冲突的配置，防止 reactor-kafka 自己反射
        props.remove(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG);
        props.remove(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG);

        // 创建 ReceiverOptions 并手动绑定实例
        ReceiverOptions<String, Object> options = ReceiverOptions.<String, Object>create(props)
            .withKeyDeserializer(new StringDeserializer())
            .withValueDeserializer(valueDeserializer);
        log.info("ReceiverOptions<Object, Object>={}", options);
        return options;
    }

    /**
     * 创建 ReactiveKafkaConsumerTemplate
     *
     * @param receiverOptions
     * @return ReactiveKafkaConsumerTemplate<String, Object>
     */
    @Bean
    public ReactiveKafkaConsumerTemplate<String, Object> reactiveKafkaConsumerTemplate(ReceiverOptions<String, Object> receiverOptions) {
        final ReactiveKafkaConsumerTemplate<String, Object> reactiveKafkaConsumerTemplate = new ReactiveKafkaConsumerTemplate<>(
            receiverOptions);
        log.info("ReactiveKafkaConsumerTemplate<String, Object>={}", reactiveKafkaConsumerTemplate);
        return reactiveKafkaConsumerTemplate;
    }

    /**
     * Json序列化器.
     *
     * @return JsonSerializer<Object>
     */
    @Bean
    public JsonSerializer<Object> jsonSerializer() {
        final JsonSerializer<Object> jsonSerializer = new JsonSerializer<>(objectMapper);
        log.info("JsonSerializer<Object>={}", jsonSerializer);
        return jsonSerializer;
    }

    /**
     * Json反序列化器.
     *
     * @return JsonDeserializer<Object>
     */
    @Bean
    public JsonDeserializer<Object> jsonDeserializer() {
        final JsonDeserializer<Object> jsonDeserializer = new JsonDeserializer<>(objectMapper);
        jsonDeserializer.addTrustedPackages(SymbolConstant.ASTERISK);
        jsonDeserializer.setUseTypeHeaders(true);
        log.info("JsonDeserializer<Object>={}", jsonDeserializer);
        return jsonDeserializer;
    }

}
